package industry_2024.industry_08.indicator

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.col

import java.util.Properties

object indicator04 {
  def main(args: Array[String]): Unit = {
    /*
          编写Scala代码，使用Spark根据dwd层的fact_change_record表关联dim_machine表统计每个车间中所有设备运行时长（即设备状态为“运行”）的
          中位数在哪个设备（为偶数时，两条数据原样保留输出），若某个设备运行状态当前未结束（即change_end_time值为空）则该状态不参与计算，计算结果
          存入MySQL数据库shtd_industry的machine_running_median表中（表结构如下），

          逻辑:就是首先求出每个设备的运行时间，然后根据车间分组，求所有设备运行时间的中位数(就是中位数位于哪个设备的上面，所以需要使用每个设备的
          运行时间作为percentile()函数的参数)
     */


    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("指标计算第四题")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
//      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
//      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    spark.table("dwd08.dim_machine").createOrReplaceTempView("dim_machine")
    spark.table("dwd08.fact_change_record")
      .where(col("changerecordstate")==="运行")
      .createOrReplaceTempView("change_record")



//  求出每个设备的运行时间
    spark.sql(
      """
        |select distinct
        |changemachineid as machine_id,
        |sum( unix_timestamp(changeendtime) - unix_timestamp(changestarttime) )
        |over(partition by changemachineid)  as time_run
        |from change_record
        |""".stripMargin).createOrReplaceTempView("r1")

    //  求每个车间，所有设备运行时间的中位数(就是中位数位于哪个设备的上面，所以需要使用每个设备的运行时间作为percentile()函数的参数)
    val result=spark.sql(
      """
        |select
        |r1.machine_id,
        |dim.machinefactory as machine_factory,
        |percentile(r1.time_run,0.5) as total_running_time
        |from r1
        |join dim_machine as dim
        |on dim.basemachineid=r1.machine_id
        |group by dim.machinefactory,r1.machine_id
        |""".stripMargin)

    result.show

    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")

    result.write.mode("overwrite")
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_industry?useSSL=false","machine_running_median08",connect)


    spark.close()
  }

}
